package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TrafficPageViewBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author Felix
 * @date 2023/3/1
 * 流量域：版本、渠道、地区、新老访客聚合统计
 */
public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的页面日志中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_vc_ch_ar_isnew_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS
            = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中的数据进行类型转换   jsonStr->jsonObj
        /*SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                return jsonObj;
            }
        });
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
            jsonStr->JSON.parseObject(jsonStr)
        );
        */
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // jsonObjDS.print(">>>");
        
        //TODO 5.按照mid进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        
        //TODO 6.为度量值进行计数   并将结果封装为一个实体类对象
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = keyedDS.process(
            new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                //声明状态  用于存放当前设备上次访问日期
                private ValueState<String> lastVisitDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> valueStateDescriptor
                        = new ValueStateDescriptor<String>("lastVisitDateState",String.class);
                    //设置状态的TTL
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");

                    String vc = commonJsonObj.getString("vc");
                    String ch = commonJsonObj.getString("ch");
                    String ar = commonJsonObj.getString("ar");
                    String isNew = commonJsonObj.getString("is_new");


                    //判断是否为独立访客
                    Long uvCt = 0L;
                    //从状态中获取当前设备上次访问日期
                    String lastVisitDate = lastVisitDateState.value();
                    //获取当前访问日期
                    Long ts = jsonObj.getLong("ts");
                    String curVisitDate = DateFormatUtil.toDate(ts);

                    if(StringUtils.isEmpty(lastVisitDate)||!lastVisitDate.equals(curVisitDate)){
                        uvCt = 1L;
                        //将当前日期更新到状态中
                        lastVisitDateState.update(curVisitDate);
                    }

                    //判断是否为新的会话
                    String lastPageId = pageJsonObj.getString("last_page_id");
                    Long svCt = StringUtils.isEmpty(lastPageId)?1L:0L;

                    //pv的计数
                    Long pvCt = 1L;

                    //持续访问时间计数
                    Long duringTime = pageJsonObj.getLong("during_time");

                    TrafficPageViewBean viewBean = new TrafficPageViewBean(
                        "",
                        "",
                        vc,
                        ch,
                        ar,
                        isNew,
                        uvCt,
                        svCt,
                        pvCt,
                        duringTime,
                        ts
                    );
                    out.collect(viewBean);
                }
            }
        );
        // beanDS.print(">>");
        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TrafficPageViewBean>() {
                        @Override
                        public long extractTimestamp(TrafficPageViewBean viewBean, long recordTimestamp) {
                            return viewBean.getTs();
                        }
                    }
                )
        );

        //TODO 8.分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedDimDS = withWatermarkDS.keyBy(
            new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                @Override
                public Tuple4<String, String, String, String> getKey(TrafficPageViewBean viewBean) throws Exception {
                    return Tuple4.of(
                        viewBean.getVc(),
                        viewBean.getCh(),
                        viewBean.getAr(),
                        viewBean.getIsNew()
                    );
                }
            }
        );

        //TODO 9.开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS
            = keyedDimDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 10.聚合计算
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TrafficPageViewBean>() {
                @Override
                public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                    value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                    value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                    value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                    value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                    return value1;
                }
            },
            new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                @Override
                public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                    for (TrafficPageViewBean viewBean : input) {
                        viewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        viewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        viewBean.setTs(System.currentTimeMillis());
                        out.collect(viewBean);
                    }
                }
            }
        );
        reduceDS.print(">>>>");
        //TODO 11.将聚合的结果写到CK中
        reduceDS.addSink(
            MyClickhouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}
